package cn.monkey.transport.mq.redisson;

import org.redisson.api.RBlockingQueue;
import org.redisson.api.RFuture;
import org.redisson.api.RedissonClient;
import org.redisson.client.codec.Codec;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

public class SimpleRedissonOutbound implements RedissonOutbound {
    private final String toTopic;
    private final RedissonClient redissonClient;
    private final RBlockingQueue<Object> blockQueue;
    private final BlockingQueue<Object> cache;
    private final Codec codec;
    private final int cacheSize;
    private final AtomicBoolean closed;

    public SimpleRedissonOutbound(String toTopic,
                                  RedissonClient redissonClient,
                                  Codec codec,
                                  int cacheSize) {
        this.toTopic = toTopic;
        this.redissonClient = redissonClient;
        this.codec = codec;
        this.cacheSize = cacheSize;
        this.cache = new ArrayBlockingQueue<>(this.cacheSize);
        this.blockQueue = this.codec == null ? redissonClient.getBlockingQueue(toTopic) : redissonClient.getBlockingQueue(toTopic, this.codec);
        this.closed = new AtomicBoolean(false);
    }

    void validClosed() {
        if (this.closed.get()) {
            throw new UnsupportedOperationException("outbound is closed");
        }
    }

    @Override
    public Mono<Void> write(Object o) {
        validClosed();
        boolean offer = this.cache.offer(o);
        return offer ? Mono.empty() : Mono.error(new UnsupportedOperationException("cache is full"));
    }

    @Override
    public Mono<Void> writeAndFlush(Object o) {
        validClosed();
        RFuture<Void> future = this.blockQueue.putAsync(o);
        return Mono.fromCompletionStage(future);
    }

    @Override
    public Mono<Void> flush() {
        validClosed();
        if (this.cache.isEmpty()) {
            return Mono.empty();
        }
        List<Object> objects = new ArrayList<>(this.blockQueue.size());
        this.cache.drainTo(objects);
        List<CompletableFuture<Void>> list = objects.stream().map(this.blockQueue::putAsync)
                .map(RFuture::toCompletableFuture)
                .toList();
        return Flux.merge(list.stream().map(Mono::fromCompletionStage).toList()).then();
    }

    @Override
    public String topic() {
        return this.toTopic;
    }

    @Override
    public Mono<Void> close() {
        this.closed.set(true);
        RFuture<Boolean> future = this.blockQueue.deleteAsync();
        return Mono.fromCompletionStage(future).then();
    }

    @Override
    public Codec codec() {
        return this.codec;
    }

    @Override
    public RedissonClient redissonClient() {
        return this.redissonClient;
    }

    @Override
    public int cacheSize() {
        return this.cacheSize;
    }
}
